package com.imooc.spark.kafka;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.server.QuotaType;

import java.util.Properties;

/**
 * kafka 生产者
 */
public class KafkaProducer extends Thread{

    private String topic;

    private Producer<Integer,String> producer;

    /*
    创建一个Producer,外面传入要写入的哪个 topic

    设置三个属性

    需要properties来创建一个ProducerConfig
    再要ProducerConfig创建一个Producer<Integer, String>
     */
    public KafkaProducer(String topic){
        this.topic = topic;

        System.out.println("要操作的topic 是："+topic);
        Properties properties = new Properties();
        properties.put("metadata.broker.list",KafkaProperties.BROKER_LIST);
        properties.put("serializer.class","kafka.serializer.StringEncoder");
        properties.put("request.required.acks","1");  //设置应答机制   0，1，-1
        producer = new Producer<Integer, String>( new ProducerConfig(properties));
    }


    //定义run方法， 来模仿消息发送
    @Override
    public void run() {

        int messageNo = 1;

        for (int i=0; i<50 ; i++){
            String message = " message : "+messageNo+" success！！" ;
            producer.send(new KeyedMessage<Integer, String>(topic,message));
            System.out.println("Sent:"+message);
            messageNo++;

            try{
                //睡两秒
                Thread.sleep(2000);
            }catch (Exception e){
                e.printStackTrace();
            }
        }

        producer.close();

    }
}
